package com.sure0000.basicKafka.example;

import org.apache.kafka.clients.admin.CreateTopicsOptions;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;

/**
 * @program: demo
 * @description: 测试 kafka topic相关操作
 * @author: xuyouchang@supcon.com
 * @create: 2019-11-13 14:24
 **/
public class KafkaTopic {

    private static final Logger logger = LoggerFactory.getLogger(KafkaTopic.class);

    public void createTopic(String bootstrap, String topic, int partitions, short replicas) throws InterruptedException {
        NewTopic newTopic = new NewTopic(topic, partitions, replicas);
        KafkaClient client = new KafkaClient(getProps(bootstrap));
        CreateTopicsOptions options = new CreateTopicsOptions();
        options.timeoutMs(1000);
        CreateTopicsResult result = client.getAdminClient().createTopics(Collections.singleton(newTopic), options);
        while (!result.all().isDone()) {
            // 等待创建完成返回信号
            Thread.sleep(500);
        }
        logger.info("is done ?", result.all().isDone());
    }

    public void listTopic(String bootstrap ) throws ExecutionException, InterruptedException {
        KafkaClient client = new KafkaClient(getProps(bootstrap));
        Set<String> topics = client.getAdminClient().listTopics().names().get();
        logger.info(topics.toString());
    }

    private Properties getProps(String bootstrap) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", bootstrap);
        return properties;
    }
}
